package com.sc.nft.admin.consume;

import com.sc.nft.dto.OpenRedPackMessageDTO;
import com.sc.nft.enums.MessageTopic;
import com.sc.nft.enums.RedisKeyEnum;
import com.sc.nft.exception.GlobalRunTimeException;
import com.sc.nft.service.EcologyPoolRedPackReceiveService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.concurrent.TimeUnit;

@RocketMQMessageListener(topic = MessageTopic.ECOLOGY_POOL_OPEN_READ_PACK + "${spring.profiles.active}", consumerGroup = MessageTopic.ECOLOGY_POOL_OPEN_READ_PACK + "${spring.profiles.active}", selectorExpression = "ECOLOGY_POOL_OPEN_RED_PACK")
@Component
@Slf4j
@RequiredArgsConstructor
public class OpenRedPackConsumer implements RocketMQListener<OpenRedPackMessageDTO> {
    private final Redisson redisson;
    private final EcologyPoolRedPackReceiveService ecologyPoolRedPackReceiveService;

    @Override
    public void onMessage(OpenRedPackMessageDTO message) {
        log.info("open red pack receive message: {}", message);
        RLock lock = redisson.getLock(String.format(RedisKeyEnum.RED_PACK_OPEN_CONSUMER.getKey(), message.getRedPackId()));
        try {
            if (!lock.tryLock(3, TimeUnit.MINUTES)) {
                throw new GlobalRunTimeException("活动红包消费mq消息失败");
            }
            ecologyPoolRedPackReceiveService.distributeConsumptionPoints(message);
        } catch (Exception e) {
            throw new GlobalRunTimeException(e.getMessage());
        } finally {
            if (lock.isLocked() && lock.isHeldByCurrentThread()) {
                lock.unlock();
            }

        }
    }
}
